package team.bluepen.supermarket.data.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import team.bluepen.supermarket.constant.Constant;
import team.bluepen.supermarket.constant.Error;
import team.bluepen.supermarket.data.entity.PathPack;
import team.bluepen.supermarket.util.Log;

import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;

/**
 * 管理Hadoop中的文件
 * @author Kai
 */
public class HdfsManager {
    private static final String TAG = "HdfsManager";
    private final Configuration conf;
    private FileSystem fileSystem;

    public HdfsManager(String url) {
        conf = new Configuration();
        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        try {
            fileSystem = FileSystem.get(URI.create(url), conf);
        } catch (IOException e) {
            fileSystem = null;
            Log.e("HDFS", e);
        }
    }

    /**
     * 上传至HDFS <br>
     * @param sourcePath 服务器源路径
     * @param destPath Hdfs目标路径
     * @return {@link Error}
     */
    public Error upload(String sourcePath, String destPath) {
        if (fileSystem == null) {
            return Error.ERROR_NULL;
        }

        Path sp = new Path(sourcePath);
        Path dp = new Path(destPath);
        try {
            fileSystem.create(dp);
            fileSystem.copyFromLocalFile(sp, dp);
        } catch (IOException e) {
            return Error.ERROR_IO;
        }
        return Error.SUCCESS;
    }

    /**
     * 从HDFS下载文件 <br>
     * 注意：此方法为下载到服务器内
     * @param sourcePath Hadoop源路径
     * @param destPath 目标路径
     * @return {@link Error}
     */
    public Error download(String sourcePath, String destPath) {
        Path sp = new Path(sourcePath);
        if (fileSystem == null) {
            return Error.ERROR_NULL;
        }
        try {
            if (!fileSystem.exists(sp)) {
                return Error.ERROR_FILE_NOT_FOUND;
            }
            FSDataInputStream fsDataInputStream = fileSystem.open(sp);
            FileOutputStream outputStream = new FileOutputStream(destPath);
            IOUtils.copyBytes(fsDataInputStream, outputStream, conf);
            outputStream.close();
        } catch (IOException e) {
            return Error.ERROR_IO;
        }
        return Error.SUCCESS;
    }

    /**
     * 从HDFS拷贝输出流 <br>
     * @param sourcePath Hadoop源路径
     * @param stream 输出流
     * @return {@link Error}
     */
    public Error download(String sourcePath, OutputStream stream) {
        Path sp = new Path(sourcePath);
        if (fileSystem == null) {
            return Error.ERROR_NULL;
        }
        try {
            if (!fileSystem.exists(sp)) {
                return Error.ERROR_FILE_NOT_FOUND;
            }
            FSDataInputStream fsDataInputStream = fileSystem.open(sp);
            IOUtils.copyBytes(fsDataInputStream, stream, conf);
        } catch (IOException e) {
            return Error.ERROR_IO;
        }
        return Error.SUCCESS;
    }

    public Error delete(String destPath) {
        Path dest = new Path(destPath);
        try {
            if (!fileSystem.exists(dest)) {
                return Error.ERROR_FILE_NOT_FOUND;
            }
            fileSystem.delete(dest, true);
        } catch (IOException e) {
            Log.e(TAG, e);
            return Error.ERROR_IO;
        }
        return Error.SUCCESS;
    }

    public List<PathPack> listAllFile(String path) {
        Path dest = new Path(path);
        List<PathPack> paths = new ArrayList<>();
        try {
            if (!fileSystem.exists(dest)) {
                return paths;
            }
            RemoteIterator<LocatedFileStatus> statusRemoteIterator = fileSystem.listFiles(dest, true);
            while (statusRemoteIterator.hasNext()) {
                LocatedFileStatus status = statusRemoteIterator.next();
                status.getLen();
                PathPack pathPack = new PathPack()
                        .setPath(status.getPath().toUri().getPath())
                        .setSize(status.getLen());
                paths.add(pathPack);
            }
        } catch (IOException e) {
            Log.e(TAG, e);
        }

        return paths;
    }

    public Error mkdirs(String path) {
        Path p = new Path(path);
        try {
            fileSystem.mkdirs(p);
        } catch (IOException e) {
            Log.e(TAG, e);
            return Error.ERROR_FILE;
        }
        return Error.SUCCESS;
    }

    public static HdfsManager getInstance() {
        return Singleton.HDFS_MANAGER;
    }

    private static final class Singleton {
        static final HdfsManager HDFS_MANAGER = new HdfsManager(Constant.HADOOP_URL);
    }

}
